package com.rickie.pvuv.streaming;

import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.HashMap;
import java.util.Map;

public class CustomEsFailureHandler implements ActionRequestFailureHandler {
    private final String index;
    public CustomEsFailureHandler(String index) {
        this.index = index;
    }

    @Override
    public void onFailure(ActionRequest actionRequest, Throwable throwable, int i, RequestIndexer indexer) throws Throwable {
        if (actionRequest instanceof IndexRequest) {
            Map<String, Object> json = new HashMap<>();
            json.put("data", ((IndexRequest) actionRequest).source());

            indexer.add(
                    Requests.indexRequest()
                            .index(index)
                            .id(((IndexRequest) actionRequest).id())
                            .source(json));
        } else {
            throw new IllegalStateException("unexpected");
        }
    }
}
